package com.wuji1626.spark.cases

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object RDD_PV_Analysis {

  def main(args: Array[String]): Unit = {
    // Step1: 准备环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    // Step2: 获取原始数据：时间戳，省份，城市，用户，广告名称
    val dataRDD = sc.textFile("datas/agent.log")

    // Step3：将原始数据进行结构转换，方便统计
    // 时间戳，省份，城市，用户，广告名称
    // =>
    // ((省份, 广告名称),1)
    val mapRDD = dataRDD.map(
      line => {
        val datas = line.split(" ")
        ((datas(1), datas(4)), 1)
      }
    )

    // Step4：将转换结构后的数据，进行分组聚合
    // ((省份，广告名称),1) => ((省份，广告名称),sum)
    val reduceRDD:RDD[((String,String), Int)] = mapRDD.reduceByKey(_ + _)

    // Step5：将聚合的结果进行结构转换
    // ((省份，广告名称),sum) => (省份，(广告名称, sum))
    val newMapRDD = reduceRDD.map({
      case ((prv, ad), sum) => {
        (prv, (ad, sum))
      }
    })

    // Step6: 将转换结构后的数据根据省份进行分组
    // (省份，[(广告名称A，sum),(广告名称B，sum)])
    val groupRDD = newMapRDD.groupByKey()

    // Step7：将分组后的数据组内进行排序（降序），取前3位
    val resultRDD = groupRDD.mapValues(
      iter => {
        // Scala 集合的默认排序是升序
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(3)
      }
    )
    resultRDD.collect().foreach(println)

    // Step8：关闭
    sc.stop()

  }

}
